package com.we.flink.userlabelplatform.table;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

public class TableRiskTag {
    public static final String KFKTABLE = "fdm.hhl01_risk_tag";
    public static final String QUERYSQL =
            "select uid, type, val from " + KFKTABLE + " where type='DIVERSION'";

    public long uid;
    public String type;
    public String val;

    public TableRiskTag() {}

    public TableRiskTag(long uid, String type, String val) {
        this.uid = uid;
        this.type = type;
        this.val = val;
    }

    @Override
    public String toString() {
        return "TableRiskTag{"
                + "uid="
                + uid
                + ", type='"
                + type
                + '\''
                + ", val='"
                + val
                + '\''
                + '}';
    }

    public static TableRiskTag of(long uid, String type, String val) {
        return new TableRiskTag(uid, type, val);
    }

    public static TableRiskTag empty(long uid) {
        return new TableRiskTag(uid, "", "");
    }

    public long getUid() {
        return uid;
    }

    public String getType() {
        return type;
    }

    public String getVal() {
        return val;
    }

    /** just for nc -lk test */
    private static class TableRiskTagMapFunction
            extends RichMapFunction<String, Tuple2<Boolean, TableRiskTag>> {
        private static final int CLMLEN = 6;
        private long uid;
        private boolean binlogFlag;
        private String type;
        private String val;

        @Override
        public Tuple2<Boolean, TableRiskTag> map(String value) throws Exception {
            try {
                String[] splits = value.split("##");
                if (splits.length != CLMLEN) {
                    System.out.println(
                            "TableRiskTagMapFunction splits length eror,input: " + value);
                    return new Tuple2<>(true, TableRiskTag.empty(0L));
                }
                if (splits[5].length() > 0) {
                    binlogFlag = Boolean.parseBoolean(splits[5]);
                }
                if (splits[2].length() > 0) {
                    uid = Long.parseLong(splits[2]);
                }
                type = splits[3];
                val = splits[4];
                return new Tuple2<>(binlogFlag, TableRiskTag.of(uid, type, val));
            } catch (Exception E) {
                System.out.println("TableRiskTagMapFunction exception input: " + value);
                System.out.println(E.getMessage());
                E.printStackTrace();
            }
            return new Tuple2<>(true, TableRiskTag.empty(0L));
        }
    }
}
